package com.lc1993929.myhdfs.server;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;

/**
 * Created by LiuChang on 2017/9/1/001.
 */
public class NameNodeHandler implements Runnable {
    private Socket sc = null;
    private Map<String, Map<Long, String>> metaMap;
    private String[] datanodes = {"hdp2", "hdp3"};


    @Override
    public void run() {
        try {
            InputStream fromClientIn = sc.getInputStream();
            OutputStream toClientOut = sc.getOutputStream();

            byte[] b = new byte[1024];
            int read = fromClientIn.read(b);
            String request = new String(b, 0, read);
            System.out.println("namenode收到客户端请求：" + request);
            //客户端请求的数据格式：PUTBLOCK:/aa/a.log:1
            String split[] = request.split(":");
            String commond = split[0];
            String path = split[1];
            String blockNumb = split[2];
            if ("PUTBLOCK".equals(commond)) {
                //构造返回数据
                long blkId = System.currentTimeMillis();
                int nextInt = new Random().nextInt(2);
                String dn = datanodes[nextInt];

                //将数据响应给客户端
                toClientOut.write((blkId + ":" + dn).getBytes());
                toClientOut.flush();

                //记录元数据
                Map<Long, String> fileMeta = metaMap.get(path);
                if (fileMeta == null) {
                    Map<Long, String> fileMap = new HashMap<>();
                    fileMap.put(blkId, dn);
                    metaMap.put(path, fileMap);
                } else {
                    fileMeta.put(blkId, dn);
                }

                toClientOut.close();
                fromClientIn.close();
                sc.close();
            }


        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    public NameNodeHandler(Socket sc, Map<String, Map<Long, String>> metaMap) {
        this.sc = sc;
        this.metaMap = metaMap;
    }
}
